with
loading as(
    select
        shipment_no transfercode,
        waybill_no billcode,
        scan_site_code scansitecode,
        next_station_code nextstationcode,
        scan_time scantime,
        package_code
    from jms_dwd.dwd_tab_barscan_loading_base_hi
    where dt between concat(date_add('{{ execution_date | cst_ds }}',-12),'-00') and '{{ execution_date |  cst_hour }}'
    and to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',-12) and '{{ execution_date |  cst_ds }}'
    and shipment_no is not null
    and next_station_code is not null
    and (
        waybill_no LIKE 'JT%'
        or (waybill_no like '55%' and length(waybill_no) = 15 and cast(waybill_no as bigint) is not null) -- 15位55开头数字 百世单号
        or (length(waybill_no) = 13 and cast(waybill_no as bigint) is not null) -- 13位数字 国标单号
    )
),
unloding as(
    select
         waybill_no billcode
        ,scan_site_code scansitecode
        ,shipment_no transfercode
        ,max(scan_time) scantime
    from jms_dwd.dwd_tab_barscan_unloading_base_hi
    where dt between concat(date_add('{{ execution_date | cst_ds }}',-1),'-00') AND '{{ execution_date |  cst_hour }}'
    and to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',0) AND '{{ execution_date |  cst_ds }}'
    group by
        waybill_no
        ,scan_site_code
        ,shipment_no
),
unloading_2 as(
    select
        waybill_no billcode
         ,shipment_no transfercode
         ,concat_ws(',', collect_set(scan_site_code)) scan_site_codes
    from jms_dwd.dwd_tab_barscan_unloading_base_hi
    where dt between concat(date_add('{{ execution_date | cst_ds }}',-1),'-00') AND '{{ execution_date |  cst_hour }}'
      and to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',0) AND '{{ execution_date |  cst_ds }}'
    group by
        waybill_no
           ,shipment_no
),
x as(
    select
        transfercode,
        billcode,
        scansitecode,
        nextstationcode,
        scantime,
        package_code,
        is_arri,
        is_unloading,
        is_loading,
        arri_scantime
     from (
        select
            transfercode,
            billcode,
            scansitecode,
            coalesce(nextstationcode_transfer, nextstationcode) nextstationcode,
            scantime,
            package_code,
            if(is_arri= 1 and nextstationcode=nextstationcode_transfer,1,0) is_arri,
            if(is_unloading=1 and nextstationcode=nextstationcode_transfer, 1, 0) is_unloading,
            if(nextstationcode_transfer=nextstationcode,1,0) is_loading,
            arri_scantime
        from(
            select
                loading.transfercode,
                loading.billcode,
                loading.scansitecode,
                loading.nextstationcode,
                loading.scantime,
                loading.package_code,
                if(unloading_1.billcode is not null, 1, 0) is_arri, --是否到达（是否在目的站点卸车）
                if(array_contains(split(nvl(scan_site_codes,''),','),nextstationcode), 1,0) is_unloading, --是否卸车（是否使用同一任务号卸车）
                unloading_2.scan_site_codes,--使用相同任务号卸车的站点
                unloading_1.scantime arri_scantime --到达时间（在目的站点卸车时间）
            from loading
            left join unloding unloading_1
                on loading.billcode = unloading_1.billcode
                and loading.transfercode = unloading_1.transfercode
                and loading.nextstationcode = unloading_1.scansitecode
            left join unloading_2
                on loading.billcode = unloading_2.billcode
                and loading.transfercode = unloading_2.transfercode
        ) t1 lateral view explode(
                split(
                        if(
                                array_contains(split(nvl(scan_site_codes,''),','),nextstationcode),
                                scan_site_codes,
                                concat_ws(',', scan_site_codes, nextstationcode)
                            ), ','
                    )
            )  tmp as nextstationcode_transfer
    ) t2
),
f1 as (
      SELECT a2.*
           FROM (SELECT
                  a.datetime ,
                  a.shipment_no shipmentno,
                  a.shipment_name,
                  a.vehicleline_code vehiclelinecode,
                  a.vehicleline_name vehiclelinename,
                  a.shipment_state shipmentstate,
                  a.start_network_name,
                  a.start_network_code,
                  a.end_network_name,
                  a.end_network_code,
                  a.start_handling_type,
                  a.end_handling_type,
                  a.billing_way,
                  a.planned_departure_time planneddeparturetime,
                  a.actual_departure_time actualdeparturetime,
                  a.planned_arrival_time plannedarrivaltime,
                  a.actual_arrival_time actualarrivaltime,
                  x.billcode,
                  a.cnt,
                  x.scantime,
                  x.package_code,
                  x.is_arri,
                  x.is_unloading,
                  x.is_loading,
                  x.arri_scantime,
                  a.vehicletype_id,
                  a.vehicletype_name,
                  a.settlement_vehicle_type_id,
                  a.settlement_vehicle_type_name,
                  row_number() OVER(PARTITION BY x.transfercode, x.billcode, x.scansitecode, x.nextstationcode ORDER BY x.is_loading desc,x.scantime,x.transfercode ) rn
                   FROM (SELECT to_date(t1.planned_departure_time) datetime,
                                t1.shipment_no,
                                t1.shipment_name,
                                t1.vehicleline_code,
                                t1.vehicleline_name,
                                t1.shipment_state,
                                t2.network_name start_network_name,
                                t2.network_code start_network_code,
                                t3.network_name end_network_name,
                                t3.network_code end_network_code,
                                t2.handling_type start_handling_type,
                                t3.handling_type end_handling_type,
                                t2.planned_departure_time,
                                t2.actual_departure_time,
                                t3.planned_arrival_time,
                                t3.actual_arrival_time,
                                t1.billing_way,
                                t1.cnt,
                                t1.vehicletype_id,
                                t1.vehicletype_name,
                                t1.settlement_vehicle_type_id,
                                t1.settlement_vehicle_type_name
                           FROM (select * from jms_dwd.dwd_tmsnew_shipment_union_base_hi where dt between '{{ execution_date | cst_ds }}' and '{{ execution_date |  cst_ds }}' and table_type=1) t1
                           LEFT JOIN (select * from jms_dwd.dwd_tmsnew_shipment_stop_union_base_hi where dt='{{ execution_date |  cst_ds }}' and table_type=1) t2
                             ON t1.shipment_no = t2.shipment_no
                           LEFT JOIN (select * from jms_dwd.dwd_tmsnew_shipment_stop_union_base_hi where dt='{{ execution_date |  cst_ds }}' and table_type=1) t3
                             ON t3.shipment_no = t2.shipment_no
                          WHERE t2.sort_num < t3.sort_num
                            AND t1.planned_departure_time BETWEEN '{{ execution_date | date_add(-7) | cst_ds }} 00:00:00' AND
                                '{{ execution_date |  cst_ds }} 23:59:59'
                             AND to_date(t1.planned_departure_time)>='2023-03-01' --todo 临时日期限制
                            ) a --核心数据源：干线线路段
                    LEFT JOIN  x
                       ON a.shipment_no = x.transfercode
                      AND a.start_network_code = x.scansitecode
                      AND a.end_network_code = x.nextstationcode
                     ) a2
          WHERE rn = 1
),
f2 as (
  select * from
  (
  select
     waybill_no as billcode
    ,package_inside_charge_weight
    ,row_number() OVER (PARTITION BY waybill_no ORDER BY dt DESC) AS rk
    ,dt
     FROM jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
       WHERE dt between date_add('{{ execution_date | cst_ds }}',-13) AND date_add('{{ execution_date |  cst_ds }}',0)
         and to_date(input_time) between date_add('{{ execution_date | cst_ds }}',-13) AND date_add('{{ execution_date |  cst_ds }}',0)
       ) way where rk = 1
),
f3 as (
    select billcode, weight from (
       SELECT
            waybill_no billcode, weight , row_number() over(partition by waybill_no order by scan_time desc) as rank
             FROM jms_dwd.dwd_tab_barscan_taking_base_hi
            WHERE weight IS NOT NULL
              and dt between concat(date_add('{{ execution_date | cst_ds }}',-13),'-00') AND '{{ execution_date |  cst_hour }}'
              AND to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',-13) AND '{{ execution_date |  cst_ds }}'
        ) tab where rank = 1
),
f4 as (
    select billcode, weight from (
       SELECT
            waybill_no billcode, weight , row_number() over(partition by waybill_no order by scan_time desc) as rank
             FROM jms_dwd.dwd_tab_barscan_warehousing_base_hi
            WHERE weight IS NOT NULL
              and dt between concat(date_add('{{ execution_date | cst_ds }}',-13),'-00') AND '{{ execution_date |  cst_hour }}'
              AND to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',-13) AND '{{ execution_date |  cst_ds }}'
        ) tab where rank = 1
),
f5 as (
    select billcode, weight from (
       SELECT
            waybill_no billcode, weight , row_number() over(partition by waybill_no order by scan_time desc) as rank
             FROM jms_dwd.dwd_tab_barscan_sitearrival_base_hi
            WHERE weight IS NOT NULL
              and scan_type = '集货到件'
              and weight IS NOT NULL
              and dt between concat(date_add('{{ execution_date | cst_ds }}',-11),'-00') AND '{{ execution_date |  cst_hour }}'
              AND to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',-11) AND '{{ execution_date |  cst_ds }}'
        ) tab where rank = 1
),
f6 as (
    select billcode,scansitecode,weight from (
       select
          billcode,scansitecode,weight,row_number() OVER(PARTITION BY billcode ORDER BY scantime asc) rank
          from (
             SELECT
               waybill_no billcode,
               scan_site_code scansitecode,
               weight,
               scan_time scantime,
               row_number() OVER(PARTITION BY waybill_no,scan_site_code ORDER BY scan_time desc) rn
             FROM jms_dwd.dwd_tab_barscan_centerarrival_base_hi
            WHERE dt between concat(date_add('{{ execution_date | cst_ds }}',-11),'-00') AND '{{ execution_date |  cst_hour }}'
              AND to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',-11) AND '{{ execution_date |  cst_ds }}'
              ) tab where tab.rn=1
        ) tmp where tmp.rank = 1
),
f7 as (
    select billcode,scansitecode,weight from (
       select
          billcode,scansitecode,weight,row_number() OVER(PARTITION BY billcode ORDER BY scantime asc) rank
          from (
             SELECT
               waybill_no billcode,
               scan_site_code scansitecode,
               weight,
               scan_time scantime,
               row_number() OVER(PARTITION BY waybill_no,scan_site_code ORDER BY scan_time desc) rn
             FROM jms_dwd.dwd_tab_barscan_centersend_base_hi
            WHERE dt between concat(date_add('{{ execution_date | cst_ds }}',-11),'-00') AND '{{ execution_date |  cst_hour }}'
              AND to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',-11) AND '{{ execution_date |  cst_ds }}'
              ) tab where tab.rn=1
        ) tmp where tmp.rank = 1
),
f8 as(
select
         waybill_no billcode
        ,scan_site_code scansitecode
        ,shipment_no transfercode
        ,max(scan_time) scantime
    from jms_dwd.dwd_tab_barscan_unloading_base_hi
    where dt between concat(date_add('{{ execution_date | cst_ds }}',-1),'-00') AND '{{ execution_date |  cst_hour }}'
    and to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',0) AND '{{ execution_date |  cst_ds }}'
    group by waybill_no
             ,scan_site_code
             ,shipment_no
),
f10 as(
select billcode
       ,scansitecode
       ,shipment_info.transfercode transfercode
       ,shipment_info.shipment_name transfername
from(
    select
             waybill_no billcode
            ,scan_site_code scansitecode
            ,max(
                struct(
                    IF(tab_barscan_unloading.shipment_no is not null and tab_barscan_unloading.shipment_no <> '', 1, 0) as flag,
                    scan_time as scantime,
                    tab_barscan_unloading.shipment_no as transfercode,
                    shipment_name
                )
            ) shipment_info
        from jms_dwd.dwd_tab_barscan_unloading_base_hi tab_barscan_unloading
        left join (select shipment_no,shipment_name from jms_dwd.dwd_tmsnew_shipment_union_base_hi where dt between '{{ execution_date | date_add(-30) | cst_ds }}' and '{{ execution_date |  cst_ds }}' and table_type=1) t
            on tab_barscan_unloading.shipment_no=t.shipment_no
        where dt between concat(date_add('{{ execution_date | cst_ds }}',-1),'-00') AND '{{ execution_date |  cst_hour }}'
        and to_date(scan_time) between date_add('{{ execution_date | cst_ds }}',0) AND '{{ execution_date |  cst_ds }}'
        -- and tab_barscan_unloading.shipment_no is not null
        group by waybill_no
                 ,scan_site_code) tt
),
f9 as(
select
   f1.datetime ,
   f1.billcode ,
   f1.shipmentno ,
   f1.start_network_code ,
   f1.end_network_code   ,
   f1.planneddeparturetime ,
   f1.actualdeparturetime ,
   f1.plannedarrivaltime ,
   f1.actualarrivaltime ,
   f1.vehiclelinecode ,
   f1.vehiclelinename ,
   f1.shipmentstate ,
   case when nvl(f6.weight, 0) <> 0 then nvl(f6.weight, 0)
       else (case when nvl(f5.weight, 0) <> 0 then nvl(f5.weight, 0)
       else (case when nvl(f4.weight, 0) <> 0 then nvl(f4.weight, 0)
       else (case when nvl(f3.weight, 0) <> 0 then nvl(f3.weight, 0)
       else null end) end) end) end as bill_weight  ,
   f2.package_inside_charge_weight as waybill_weight ,
   f3.weight as taking_weight ,
   f4.weight as housing_weight ,
   f5.weight as site_weight ,
   f6.weight as firstarrival_weight ,
   f7.weight as firstsend_weight ,
   f1.start_network_name ,
   f1.end_network_name   ,
   f1.start_handling_type ,
   f1.end_handling_type   ,
   f1.shipment_name       ,
   f1.billing_way         ,
   f1.cnt                 ,
   f1.is_arri                ,
   f1.is_unloading,
   f1.is_loading,
   from_unixtime(unix_timestamp()) as update_time    ,
   f6.scansitecode as firstarrival_code ,
   f7.scansitecode as firstsend_code,
   abs(hash(rand())) as uid,
   1 as is_delete,
   f1.datetime as dt,
   row_number() over(partition by f1.billcode,f1.start_network_code,f1.end_network_code,f1.is_arri order by f1.scantime desc) rank, --同一站点多次扫描，多次扫描下一站一致但是任务号不同
   row_number() over(partition by f1.billcode,f1.start_network_code,f1.shipmentno,f1.is_arri order by f1.arri_scantime desc nulls last) rank_2, --同一站点多次扫描，多次扫描任务号相同
    if(f10.billcode is not null,1,0) is_arri_site, --是否在站点卸车
   f10.transfercode,
   f10.transfername,
   row_number() over(partition by f1.billcode,f1.end_network_code order by f1.is_arri desc,f1.is_unloading desc, f1.scantime desc) rank_3, --在同一站点卸车，优先取在始发站点装车对应任务号 否则取最晚装车对应任务号
   collect_set(if(f1.is_arri=0 and f1.is_unloading=0 and f10.billcode is not null, f1.shipmentno, null)) over(partition by f1.billcode,f1.end_network_code) loading_shipmentno_list,--同一卸车站点，对应所有装车任务号列表
   f1.package_code,
   f1.vehicletype_id,
   f1.vehicletype_name,
   f1.settlement_vehicle_type_id,
   f1.settlement_vehicle_type_name
from f1
left join f2 on f1.billcode=f2.billcode
left join f3 on f1.billcode=f3.billcode
left join f4 on f1.billcode=f4.billcode
left join f5 on f1.billcode=f5.billcode
left join f6 on f1.billcode=f6.billcode
left join f7 on f1.billcode=f7.billcode
-- left join f8 on f1.billcode=f8.billcode and f1.end_network_code = f8.scansitecode and f1.shipmentno = nvl(f8.transfercode,f8.billcode)
left join f10 on f1.billcode=f10.billcode and f1.end_network_code = f10.scansitecode
)
insert overwrite table jms_dm.dm_shipno_weight_detail_dt partition(dt)
select
    datetime ,
    billcode ,
    shipmentno ,
    start_network_code ,
    end_network_code   ,
    planneddeparturetime ,
    actualdeparturetime ,
    plannedarrivaltime ,
    actualarrivaltime ,
    vehiclelinecode ,
    vehiclelinename ,
    shipmentstate ,
    bill_weight  ,
    waybill_weight ,
    taking_weight ,
    housing_weight ,
    site_weight ,
    firstarrival_weight ,
    firstsend_weight ,
    start_network_name ,
    end_network_name   ,
    start_handling_type ,
    end_handling_type   ,
    shipment_name       ,
    billing_way         ,
    cnt                 ,
    if(rank=1 and is_arri=1 and rank_2=1,1,0)   is_arri, --如果多次装车扫描均能匹配到卸车扫描，则取最新；如果只有一次装车能匹配到卸车扫描，则取匹配得上的数据
    update_time    ,
    firstarrival_code ,
    firstsend_code,
    uid,
    is_delete,
    if(is_loading=1 and is_arri=0 and is_unloading = 0 and is_arri_site=1,1,0) is_arri_innormal,--是否异常卸车
    if(is_loading=1 and is_arri=0 and is_unloading = 0 and is_arri_site=1,transfercode,null) shipmentno_innormal,--异常任务单号
    if(is_loading=1 and is_arri=0 and is_unloading = 0 and is_arri_site=1,transfername,null) shipment_name_innormal,--异常任务单名
    package_code,--包号
    if(
        (is_arri=0 and is_unloading = 0 and is_arri_site=1 and rank_3=1) or --未在目的站点卸车 但是在目的站点卸车 且装车任务号个数大于1
        (is_arri=0 and is_unloading=1 and rank_3=1) --未在目的站点卸车 但是使用相同任务号在其他站点卸车
    ,1,0) is_rough_judgement,--是否粗判
    if(
        (is_arri=0 and is_arri_site=1 and rank_3=1) --未在目的站点卸车 但是在目的站点卸车 且装车任务号个数大于1
        ,concat_ws('/', loading_shipmentno_list)
        ,null
    ) compare_shipment_list,
    vehicletype_id,
    vehicletype_name,
    settlement_vehicle_type_id,
    settlement_vehicle_type_name,
    is_loading,
    dt
from f9
where billing_way in (1,2)
and shipmentstate <> 5
and datetime = '{{ execution_date |  cst_ds }}'
 distribute by dt,pmod(hash(rand()), 10)
;